package com.heima.article.listener;

import com.alibaba.fastjson.JSON;
import com.heima.article.dto.ArticleStreamMessage;
import com.heima.article.dto.UpdateArticleMessage;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.util.StringUtils;

import java.time.Duration;

@EnableBinding(value = ArticleProcess.class)
public class ArticleListener {
    @Value("${commit.time}")
    private String time;
    //指定接收消息的主题
    @StreamListener(value = "article_behavior")
    //指定发送结果的主题
    @SendTo(value = "article_result")
    public KStream<String,String> process(KStream<String,String> input){
        //接收到的消息格式为UpdateArticleMessage
        //解析json，提取文章id  UpdateArticleMessage
        KStream<String, String> map = input.map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(String key, String value) {
                UpdateArticleMessage updateArticleMessage = JSON.parseObject(value, UpdateArticleMessage.class);
                Long articleId = updateArticleMessage.getArticleId();
                return new KeyValue<>(articleId.toString(), value);
            }
        });
        //根据每一篇文章的id来进行分组
        KGroupedStream<String, String> stringStringKGroupedStream = map.groupByKey();
        //统计时间窗口的数据
        TimeWindowedKStream<String, String> stringStringTimeWindowedKStream = stringStringKGroupedStream.windowedBy
                (TimeWindows.of(Duration.ofMillis(Long.parseLong(time))));
        //进行聚合处理
        Initializer<String> init = new Initializer<String>() {
            @Override
            public String apply() {
                //在这个时间窗口内 第一次进来的数据
                return null;
            }
        };
        Aggregator<String,String, String> aggr  = new Aggregator<String, String, String>() {
            @Override
            public String apply(String key, String value, String aggregate) {
                ArticleStreamMessage articleStreamMessage; // 结果类对象
                UpdateArticleMessage updateArticleMessage = JSON.parseObject(value, UpdateArticleMessage.class);
                Long articleId = updateArticleMessage.getArticleId();
                if (StringUtils.isEmpty(aggregate)){
                    //ArticleStreamMessage
                    articleStreamMessage = new ArticleStreamMessage();
                    articleStreamMessage.setArticleId(articleId);
                }else {
                    ArticleStreamMessage articleStreamMessage1 = JSON.parseObject(aggregate, ArticleStreamMessage.class);
                    articleStreamMessage = articleStreamMessage1;
                    articleStreamMessage.setArticleId(articleId);
                }
                UpdateArticleMessage updateArticleMessage1 = JSON.parseObject(value, UpdateArticleMessage.class);
                switch (updateArticleMessage1.getType()){
                    //操作类型 0 阅读 1 点赞 2 评论 3 收藏
                    case 0 :
                        articleStreamMessage.setView(articleStreamMessage.getView() + updateArticleMessage1.getAdd());
                        break;
                    case 1:
                        articleStreamMessage.setLike(articleStreamMessage.getLike() + updateArticleMessage1.getAdd());
                        break;
                    case 2:
                        articleStreamMessage.setCollect(articleStreamMessage.getCollect() + updateArticleMessage1.getAdd());
                        break;
                    case 3:
                        articleStreamMessage.setComment(articleStreamMessage.getComment() + updateArticleMessage1.getAdd());
                        break;
                }
                String s = JSON.toJSONString(articleStreamMessage);
                return s;
            }
        };
        //聚合处理的结果是 ArticleStreamMessage
        //最终发送到结果的value是ArticleStreamMessage转换成json格式
        KTable<Windowed<String>, String> aggregate = stringStringTimeWindowedKStream.aggregate(init, aggr);
        KStream<Windowed<String>, String> windowedStringKStream = aggregate.toStream();
        KStream<String, String> map1 = windowedStringKStream.map(new KeyValueMapper<Windowed<String>, String, KeyValue<String,String>>() {
            @Override
            public KeyValue<String, String> apply(Windowed<String> key, String value) {
                return new KeyValue<>(key.key(), value);
            }
        });
        return map1; // 发送聚合结果
    }
}
